package com.talent.core.protocol.mqtt.handler;

import com.talent.core.protocol.mqtt.TlMqttSession;
import com.talent.core.protocol.mqtt.config.Constant;
import com.talent.core.protocol.mqtt.enums.TlMqttMessageType;
import com.talent.core.protocol.mqtt.enums.TlMqttQos;
import com.talent.core.protocol.mqtt.model.entity.TlMessage;
import com.talent.core.protocol.mqtt.model.entity.TlSubClient;
import com.talent.core.protocol.mqtt.model.fix.TlMqttFixedHead;
import com.talent.core.protocol.mqtt.model.payload.TlMqttPublishPayload;
import com.talent.core.protocol.mqtt.model.request.TlMqttPublishReq;
import com.talent.core.protocol.mqtt.model.response.TlMqttPubAck;
import com.talent.core.protocol.mqtt.model.response.TlMqttPubRec;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPubAckVariableHead;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPubRecVariableHead;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPublishVariableHead;
import com.talent.core.protocol.mqtt.store.IMqttStore;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Set;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
public class TlMqttPublishHandler extends SimpleChannelInboundHandler<TlMqttPublishReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttPublishHandler.class);

    private final IMqttStore mqttStore;


    public TlMqttPublishHandler(IMqttStore mqttStore){
        this.mqttStore = mqttStore;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttPublishReq req) throws Exception {

        logger.debug("进入【PUBLISH】处理器");
        String clientId = ctx.channel().attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get().toString();
        TlMqttFixedHead fixedHead = req.getFixedHead();
        TlMqttPublishVariableHead variableHead = req.getVariableHead();
        TlMqttPublishPayload payload = req.getPayload();

        int retain = fixedHead.getRetain();
        int messageQos = fixedHead.getQos();
        String topic = variableHead.getTopic();
        Integer messageId = variableHead.getMessageId();
        String content = payload.getContent().toString();
        logger.info("收到客户端{}向主题{}发送的消息{}",clientId,topic,content);

        /*如果是保留消息 存储*/
        if (retain == 1) {
            storeRetain(topic, content);
        }

        TlMqttQos messageQoS = TlMqttQos.valueOf(messageQos);
        /**收到消息后恢复客户端*/
        switch (messageQoS){
            case QOS1:
                sendAck(messageId,ctx);
                break;
            case QOS2:
                sendRec(messageId,ctx,req);
                break;
            default:
                break;

        }
        /*如果发送的消息是qos2的话就直接跳过不传递*/
        if(messageQoS==TlMqttQos.QOS2){
            return;
        }

        /*获取到订阅这个主题的所有客户端*/
        List<TlSubClient> subClients = mqttStore.getSubClientByTopic(topic);
        for (TlSubClient subClient : subClients) {

            int subQos = subClient.getQos();
            logger.info("向客户端客户端{}订阅主题{}的qos是{},发布消息的{}",subClient.getClientId(),subClient.getTopic(),subQos,messageQos);
            /*消息的订阅级别取最小的*/
            int readQos= Math.min(subQos, messageQos);
            TlMqttQos mqttQoS = TlMqttQos.valueOf(readQos);
            TlMqttSession session = subClient.getSession();

            switch (mqttQoS){
                case QOS0:
                    int length = fixedHead.getLength();
                    TlMqttQos tlMqttQos = TlMqttQos.valueOf(messageQos);
                    if(tlMqttQos==TlMqttQos.QOS1||tlMqttQos==TlMqttQos.QOS2){
                        length-=Constant.SHORT_LENGTH;
                    }
                    forward(session,topic,mqttQoS,retain,content,length);
                    break;
                case QOS1:
                    TlMqttPublishReq publishReq = forward(session, topic, mqttQoS, retain, content, fixedHead.getLength());
                    /*创建一个定时任务 判断是否收到了ack消息 如果没有就继续发送这 todo*/
                    break;
                default:
                    break;

            }
        }


    }


    /**
     * 保存保留消息 在新订阅的时候发送
     * @param topic 主题
     * @param content 内容
     */
    private void storeRetain(String topic,String content) {

        int remindLength = Constant.SHORT_LENGTH + topic.getBytes().length
                + Constant.SHORT_LENGTH + content.getBytes().length;
        TlMqttPublishReq pubMsg = new TlMqttPublishReq(topic,
                TlMqttQos.QOS0, 0, 1,
                content, null, remindLength);
        TlMessage message = new TlMessage();
        message.setReq(pubMsg);
        message.setTopic(topic);
        message.setTimestamp(System.currentTimeMillis());
        mqttStore.putRetainMessage(topic, message);
    }


    /**
     * 向其他客户端转发消息
     * @param session 会话
     * @param topic 主题
     * @param mqttQos 消息等级
     * @param retain 是否是保留消息
     * @param content 内容
     * @param length 数据长度
     * @return 转发给其他客户端的消息
     */
    private TlMqttPublishReq forward(TlMqttSession session , String topic, TlMqttQos mqttQos, int retain, String content, int length){
        Integer identifierId = mqttStore.getMessageId();

        //如果
        if(mqttQos==TlMqttQos.QOS0){
            identifierId=null;
        }
        TlMqttPublishReq publishReq = new TlMqttPublishReq(topic, mqttQos, 0, retain,
                content,  identifierId, length);
        ChannelHandlerContext ctx = session.getCtx();
        //如果转发的消息是qos1的并且是保留会话
        if(mqttQos==TlMqttQos.QOS1){
            mqttStore.brokerStoreMsg(identifierId,publishReq);
        }

        //存储消息
        Set<TlMqttPublishReq> republish = session.getRepublish();
        republish.add(publishReq);

        if(null!=ctx){
            ctx.channel().writeAndFlush(publishReq);
        }
        return publishReq;
    }

    /**
     * 用于重发消息
     * @param session 会话
     * @param publishReq 消息
     * @return 返回消息
     */
    private TlMqttPublishReq forward(TlMqttSession session,TlMqttPublishReq publishReq){

        ChannelHandlerContext ctx1 = session.getCtx();
        if(null!=ctx1){
            //设置重复消息
            publishReq.getFixedHead().setDup(1);
            ctx1.channel().writeAndFlush(publishReq);
        }
        return publishReq;
    }

    /**
     * 构建ack消息发送返回
     * @param messageId 消息ID
     * @param ctx 通道
     */
    private void sendAck(int messageId, ChannelHandlerContext ctx){
        TlMqttPubAck res = new TlMqttPubAck();
        TlMqttFixedHead head = new TlMqttFixedHead();
        head.setMessageType(TlMqttMessageType.PUBACK.getValue());
        head.setLength(Constant.SHORT_LENGTH);
        TlMqttPubAckVariableHead ackVariableHead = new TlMqttPubAckVariableHead();
        ackVariableHead.setMessageId(messageId);
        res.setFixedHead(head);
        res.setVariableHead(ackVariableHead);
        ctx.channel().writeAndFlush(res);
    }

    private void sendRec(int messageId,ChannelHandlerContext ctx ,TlMqttPublishReq req){
        //如果这个包在存在就说嘛时一个重复报 直接丢弃
        if(mqttStore.getStoreMsg(messageId)!=null){
            return;
        }
        //本地存起来
        mqttStore.brokerStoreMsg(messageId,req);
        //发送rec消息给发送者
        TlMqttFixedHead head = new TlMqttFixedHead();
        head.setMessageType(TlMqttMessageType.PUBREC.getValue());
        head.setLength(Constant.SHORT_LENGTH);
        TlMqttPubRecVariableHead recVariableHead = new TlMqttPubRecVariableHead();
        recVariableHead.setMessageId(messageId);
        TlMqttPubRec res = new TlMqttPubRec(head, recVariableHead);
        ctx.channel().writeAndFlush(res);
    }
}
